package com.dgz.wc;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: DongGuoZhen
 * @Date: 2025/08/20/14:13
 * @Description:
 */
// DataSetApi 批处理 不推荐
public class WordCountBatchDemo {

    public static void main(String[] args) {
//        1. 创建一个执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
//        2. 读取数据 从文件中读取
        DataSource<String> lineDs = env.readTextFile("input/word.txt");
//        3. 按行切分，转换成(word,1)
        FlatMapOperator<String, Tuple2<String, Integer>> wordAndOne = lineDs.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
//                3.1 按照空格切分单词
                String[] words = s.split(" ");
//                3.2 将单词转换为word 1
                for (String word : words) {
                    Tuple2<String, Integer> wordTuple2 = Tuple2.of(word, 1);
//                    3.3 使用collector向下游发送数据
                    collector.collect(wordTuple2);
                }
            }
        });
//        4. 分组
        UnsortedGrouping<Tuple2<String, Integer>> wordAndOneGroupBy = wordAndOne.groupBy(0); // 0 表示第一个字段
//        5. 各分组内聚合
        AggregateOperator<Tuple2<String, Integer>> result = wordAndOneGroupBy.sum(1); // 1 表示第二个字段
//        6. 输出
        try {
            result.print();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
